package com.huobi.pro.market.client;


import java.net.URISyntaxException;
import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
import java.util.List;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

import org.apache.commons.lang.RandomStringUtils;
import org.java_websocket.client.WebSocketClient;
import org.java_websocket.handshake.ServerHandshake;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;

import com.alibaba.fastjson.JSONObject;
import com.huobi.HuobiProperty;
import com.huobi.event.EventPublisher;
import com.huobi.pro.market.HuobiWsClient;
import com.huobi.pro.market.ZipUtil;

import lombok.extern.slf4j.Slf4j;

/**
 * 市场行情接口
 */
@Service("LinearswapClient")
@Slf4j
public class LinearswapClient implements HuobiWsClient {

    @Autowired ClientAuth clientAuth;
    @Autowired EventPublisher eventPublisher;
    @Autowired HuobiProperty property;
    private WebSocketClient webSocketClient;
    private final static String path = "/linear-swap-ws";
    private ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(2);

    public LinearswapClient() {
    }

    public void sub(List<String> channels) throws URISyntaxException {
        doConnect(channels);
    }

    private void doConnect(List<String> channels) throws URISyntaxException {

        webSocketClient = new WebSocketClient(property.getURI(path)) {

            public void onOpen(ServerHandshake serverHandshake) {
                clientAuth.sendAuth(LinearswapClient.this, path);
                doSub(channels);
                dealReconnect();
            }

            public void onMessage(String s) {
                log.debug("onMessage:{}", s);
            }

            @Async
            public void onMessage(ByteBuffer bytes) {
                try {
                    String message = new String(ZipUtil.decompress(bytes.array()), StandardCharsets.UTF_8);
                    log.info("receive message {}", message);
                    JSONObject data = JSONObject.parseObject(message);
                    String ch = data.getString("ch");
                    Object ping = data.get("ping");
                    if (ping != null) {
                    	ping(data.getLong("ping"));
                    }
                    if (ch != null) {
                    	switch (ch) {
                    		case "market.TRX-USDT.kline.1min":
                    			eventPublisher.publishKLine(data);
                    			break;
                    		case "market.TRX-USDT.trade.detail":
                    			eventPublisher.publishTrade(data);
                    			break;
                    	}
                    }
                } catch (Exception e) {
                    log.error("parse message error.", e);
                }
            }

            public void onClose(int i, String s, boolean b) {
                log.error("onClose i:{},s:{},b:{}", i, s, b);
            }

            public void onError(Exception e) {
                log.error("onError:", e);
            }
        };
        webSocketClient.connect();
    }

    private void doSub(List<String> channels) {
        channels.stream().forEach(e -> {
            JSONObject sub = new JSONObject();
            sub.put("sub", e);
            sub.put("id", RandomStringUtils.randomAlphanumeric(10));
            webSocketClient.send(sub.toString());
        });
    }

    private void dealReconnect() {
        try {
            scheduledExecutorService.scheduleAtFixedRate(() -> {
                try {
                    if ((webSocketClient.isClosed() && !webSocketClient.isClosing())) {
                        log.error("isClosed:{},isClosing:{}，准备重连", webSocketClient.isClosed(), webSocketClient.isClosing());
                        Boolean reconnectResult = webSocketClient.reconnectBlocking();
                        log.error("重连的结果为：{}", reconnectResult);
                        if (!reconnectResult) {
                            webSocketClient.closeBlocking();
                            log.error("closeBlocking");
                        }
                    }
                } catch (Throwable e) {
                    log.error("dealReconnect异常", e);
                }
            }, 60, 10, TimeUnit.SECONDS);
        } catch (Exception e) {
            log.error("dealReconnect scheduledExecutorService异常", e);
        }

    }

    @Override
    public void send(String text) {
        this.webSocketClient.send(text);
    }

    public HuobiProperty getHuoboProperty() {
        return property;
    }

    public void ping(Long timestamp) {
        JSONObject ping = new JSONObject();
        ping.put("pong", timestamp);
        send(ping.toString());
    }


}